package cn.itcast.dmp.area

import cn.itcast.dmp.area.AreaProcessor.logWarning
import cn.itcast.dmp.config.AppConfigHelper
import cn.itcast.dmp.process.Processor
import cn.itcast.dmp.utils.HttpUtils
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 依据经纬度拼接请求URL，调用高德API，解析JSON数据，获取商圈信息
  */
object AreaProcessor extends Processor {


	/**
	  * 开发步骤：
		1、获取GeoHash值、经度和维度
		2、依据经纬度范围，过滤非大陆的数据
		3、按照GeoHash去重（使用group by），获取组内经纬度平均值
		4、获取已有商圈信息
		5、与ODS表关联（左外连接），获取新的经纬度信息
		6、自定义UDF，传递经纬度，调用高德API，获取商圈信息
		7、保存商圈信息到Kudu表（GeoHash，Area）
		  */
	override def processData(odsDF: DataFrame): DataFrame = {
		// 获取SparkSession对象，导入隐式转换和函数库
		val spark: SparkSession = odsDF.sparkSession
		import spark.implicits._
		import org.apache.spark.sql.functions._

		val geoHashDF: DataFrame = odsDF
			// 1、获取GeoHash值、经度和维度
			.select($"geoHash", $"longitude", $"latitude")
			// 2、依据经纬度范围，过滤非大陆的数据: 经度73.66~135.05，纬度3.86~53.55
			.filter(
			$"longitude".gt(73.66).and($"longitude".lt(135.05))
				.and($"latitude".gt(3.86)).and($"latitude".lt(53.55))
		)
			//3、按照GeoHash去重（使用group by），获取组内经纬度平均值
			.groupBy($"geoHash").agg(
			round(avg("longitude"), 6).as("avg_longitude"),
			round(avg("latitude"), 6).as("avg_latitude")
		)
		logWarning(s"从Kudu表读取的ODS数据量: ${odsDF.count()} ...............")
		logWarning(s"将ODS表数据过滤去重以后的数据量: ${geoHashDF.count()} ...............")


		// 4、获取已有商圈信息
		import cn.itcast.dmp.untils.KuduUtils._
		val areaDFOption: Option[DataFrame] = spark.readKuduTable(AppConfigHelper.BUSINESS_AREAS_TABLE_NAME)


		// 5、与ODS表关联（左外连接），获取新的经纬度信息
		val geoHashWithAreaDF: DataFrame = areaDFOption match {
			case Some(df) =>
				// 当Kudu中存储商圈信息库表的时候，再进行关联分析，使用left连接，geoHash字段
				geoHashDF
					.join(
						df, // 关联的表
						geoHashDF.col("geoHash") === df.col("geo_hash"), // 指定关联字段
						"left" // 关联方式
					)
					// 关联以后，将未获取到值area数据保存，请求高德API获取；获取到值的 说明是重复的数据
					.filter($"area".isNull)
				case None => geoHashDF
		}
		logWarning(s"去重已存在商圈信息表的数据后数据量: ${geoHashWithAreaDF.count()} ...............")


		// 4、自定义UDF，传递经纬度，调用高德API，获取商圈息
		// 注册函数，传递经纬度获取商圈信息
		val get_area: UserDefinedFunction = udf(HttpUtils.loadGaode2Area _) // 当高阶函数中传递函数的名称的时候，需要加空格和下划线，表示传递是一个函数

		val areaDF: DataFrame = geoHashWithAreaDF.select(
			$"geoHash".as("geo_hash"),
			get_area($"avg_longitude", $"avg_latitude").as("area")
		)

		// 5、返回构建商圈库信息
		areaDF
	}


}
